package net.bwie.douyin.dwd.log1.Function;


import net.bwie.douyin.dwd.log1.bean.AlertMessageUser;
import net.bwie.douyin.dwd.log1.bean.TikTokBean;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @ClassName：realTimeMessageCountFunction
 * @Author: ZddddQ
 * @Date: 2025/5/27 22:26
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
public class realTimeMessageCountFunction implements WindowFunction<TikTokBean, AlertMessageUser, String, TimeWindow> {

    private FastDateFormat fastDateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");
    @Override
    public void apply(String s, TimeWindow window, Iterable<TikTokBean> input, Collector<AlertMessageUser> out) throws Exception {


        String windowStartTime = fastDateFormat.format(window.getStart());
        String windowEndTime = fastDateFormat.format(window.getEnd());


        int count = 0;
        for (TikTokBean bean : input) {
            count++;
        }

        if (count >= 2) {
            AlertMessageUser alert = new AlertMessageUser();
            alert.setUserId(s);
            alert.setWindowStartTime(windowStartTime);
            alert.setWindowEndTime(windowEndTime);
            alert.setMessageCount(count);
            alert.setTs(System.currentTimeMillis());
            out.collect(alert);
        }
    }
}
